package org.baojie.raft.connector;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.baojie.raft.box.Box;
import org.baojie.raft.connector.client.RaftNettyClient;
import org.baojie.raft.connector.server.RaftNettyServer;
import org.baojie.raft.message.RaftMessage;
import org.baojie.raft.serialization.RaftSerialize;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadLocalRandom;

public class RaftChannel implements Connector {

    private static final Logger log = LoggerFactory.getLogger(RaftChannel.class);

    private final ThreadLocalRandom random = ThreadLocalRandom.current();
    private final RaftSerialize serialize = RaftSerialize.getInstance();
    private final RaftNettyServer server;
    private final RaftNettyClient client;


    public RaftChannel(Box box) {
        this.server = new RaftNettyServer(32, box);
        this.client = new RaftNettyClient(64, box);
    }

    @Override
    public boolean start() {
        return server.start();
    }

    @Override
    public void sendMessage(RaftMessage message) {
        if (null == message) {
            return;
        } else {
            Channel channel = client.getChannel(message.remote());
            if (null == channel) {
                log.error("");
            } else {
                byte[] bytes = serialize.serialize(message);
                doSend(channel, bytes);
            }
        }
    }

    private void doSend(Channel channel, byte[] bytes) {
        channel.writeAndFlush(bytes).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isCancelled() || (!future.isSuccess())) {
                    log.error("");
                } else if (future.isSuccess() || future.isDone()) {
                    log.debug("");
                } else {
                    log.warn("");
                }
            }
        });
    }

    @Override
    public void stop() {
        try {
            server.stop();
        } finally {
            client.stop();
        }
    }

}
